package com.alison.datastream.chapter6_timeAndWaterMark;

import cn.hutool.core.date.DateUtil;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import org.slf4j.LoggerFactory;

import java.util.UUID;

/**
 * @Author alison
 * @Date 2024/4/4 13:09
 * @Version 1.0
 * @Description
 */

public class WatermarkTest {
    // log4j2方式
//    private static final org.apache.logging.log4j.Logger logger = org.apache.logging.log4j.LogManager.getLogger(WatermarkTest.class);
    // slf4j 方式
    private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(LoggerFactory.class);


    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置周期性水印的间隔
        env.getConfig().setAutoWatermarkInterval(5000L);


        DataStream<Tuple2<String, Long>> dataStream = env.addSource(new MySource());

        DataStream<Tuple2<String, Long>> withTimestampsAndWatermarks = dataStream.assignTimestampsAndWatermarks(new WatermarkStrategy<Tuple2<String, Long>>() {
            @Override
            public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Tuple2<String, Long>>() {
                    private long maxTimestamp = 0L;
                    private long delay = 3000; //延迟3s

                    @Override
                    public void onEvent(Tuple2<String, Long> event, long eventTimestamp, WatermarkOutput output) {
                        maxTimestamp = Math.max(maxTimestamp, event.f1);
                        System.out.println("onEvent: " + maxTimestamp);
                    }

                    // setAutoWatermarkInterval(5000L) 设置的周期提交水印， 5s一次,否则默认是Integer.MIN_VALUE
                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        //  Watermark比currentMaxTimestamp最大值慢3s, 超过3s的被视为迟到数据
                        System.out.println("onPeriodicEmit: " + (maxTimestamp - delay));
                        output.emitWatermark(new Watermark(maxTimestamp - delay));

                    }
                };
            }
        });

//使用内置的水印生成器
//		DataStream<Tuple2<String,Long>> withTimestampsAndWatermarks = dataStream.assignTimestampsAndWatermarks(
//				WatermarkStrategy
//						.<Tuple2<String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//						.withTimestampAssigner((event, timestamp)->event.f1));

        withTimestampsAndWatermarks.process(new ProcessFunction<Tuple2<String, Long>, Object>() {

            @Override
            public void processElement(Tuple2<String, Long> value, Context ctx, Collector<Object> out) throws Exception {
                long w = ctx.timerService().currentWatermark();
//                logger.info("value: {}", value.toString());
//                logger.info("thread-name: {}, 水印: {}, water date: {}, now: {} ",
//                        Thread.currentThread().getName(), w, DateUtil.formatDate(new java.util.Date(w)), DateUtil.formatDate(new java.util.Date(value.f1)));
                System.out.println(String.format("value: %s", value.toString()));
                System.out.println(String.format("%s thread-name: %s, 水印: %s, water date: %s, now: %s ", DateUtil.now(), Thread.currentThread().getName(), w, DateUtil.formatDateTime(new java.util.Date(w)), DateUtil.formatDateTime(new java.util.Date(value.f1))));
            }
        });

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class MySource implements SourceFunction<Tuple2<String, Long>> {
        private volatile boolean isRunning = true;

        @Override
        public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {
            while (isRunning) {
                Thread.sleep(1000);
                //订单id
                String orderid = UUID.randomUUID().toString().replace("-", "");
                //订单完成时间
                long orderFinishTime = System.currentTimeMillis();
                ctx.collect(Tuple2.of(orderid, orderFinishTime));
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
